#define MSEVENT_PUSH_C
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
//user header
#include <libmscommon/mscommon.h>
#include <libmscommon/msstring.h>
#include <libmscommon/mstime.h>
#include <libmscommon/msmd.h>
#include <libmscommon/msfifo.h>
#include <libmscommon/msthread.h>
#include <libmscommon/msepoll.h>
#include <libmsprotocol/msprotocol.h>
#include <libmslog/mslog.h>
#include "mseventPush.h"

#define FLAG "MSEVP"

#define MAXNUM_MSG	200
#define MAXNUM_EVP_GW 16
MSEVENTPUSHContext evp_ctt;

typedef struct MSEVENTPUSHDATAContext{
	ms_bool flag_start;
	ms_bool flag_stop;
	MSFIFOLISTContext *pFifoList_ctt;
	MSLOCKContext mslock_ctt;
}MSEVENTPUSHDATAContext;

typedef struct MSEVENTMSGContext{
	ms_u64 inTime_us;
	MSEVENTPUST_TIME timePush;
	MSEVENT_LEVEL evLevel;
	ms_byte evType[64];
	ms_byte devId[128];
	ms_byte jsSrc[1024];
}MSEVENTMSGContext;

static MSEVENTPUSHDATAContext msEvpData_ctt;

typedef struct MSEVPGWContext{
	ms_bool flag_stop;
	ms_byte name[64];
	URLContext *pUrl_ctt;
	URLProtocol *msptl_push;
}MSEVPGWContext;
typedef struct MSEVENTMSGFIFOContext{
	ms_u64 timePush_base;
	ms_s32 numMsg;
	MSEVENTMSGContext msgList_ctt[MAXNUM_MSG]; 
}MSEVENTMSGFIFOContext;

static ms_string msevpType_str[]={MSEVP_EVENTTYPE_STR};

static ms_void msevp_innerapi_buildMsg(ms_string jsOut,ms_u64 timeUs,MSEVENT_LEVEL evLevel,ms_string evType,ms_string devId,ms_string jsSrc)
{
	char strtime[64]={0};
	mstime_api_us2DTime((ms_u64 *)&timeUs,mstime_fmt_datetime_ex1,strtime);
	ms_byte tbuf[1024]={0};
	ms_strcat(jsOut, "{");
	ms_strcats(tbuf, jsOut, "\"level\":%d", evLevel);
	ms_strcats(tbuf, jsOut, ",\"time\":\"%s\"", strtime);
	if(ms_null!=evType){
		ms_strcats(tbuf, jsOut, ",\"type\":\"%s\"", evType);
	}
	if(ms_null!=devId){
		ms_strcats(tbuf, jsOut, ",\"devId\":\"%s\"", devId);
	}
	if(ms_null!=jsSrc){
		ms_strcats(tbuf, jsOut, ",\"data\":[%s]", jsSrc);
	}
	ms_strcat(jsOut, "}");

}
static ms_s32 msevp_innerapi_msgPush(MSEVPGWContext *pEVPGw_ctt,ms_s32 numMsg,MSEVENTMSGContext *pEvMsgList_ctt)
{
	ms_s32 index_realMsg=0;
	if(numMsg>0){
		ms_byte *msgBuf={0};
		ms_malloc_ret(numMsg, msgBuf, (2048*numMsg), "msgBuf");
		ms_strcat(msgBuf, "{");
		ms_strcat(msgBuf, "\"msgList\":[");
		for(index_realMsg=0;index_realMsg<numMsg;index_realMsg++){
			ms_byte msg[2048]={0};
			msevp_innerapi_buildMsg(msg, pEvMsgList_ctt[index_realMsg].inTime_us, 
				pEvMsgList_ctt[index_realMsg].evLevel, pEvMsgList_ctt[index_realMsg].evType, 
				pEvMsgList_ctt[index_realMsg].devId, pEvMsgList_ctt[index_realMsg].jsSrc);
			if(index_realMsg>0){
				ms_strcat(msgBuf, ",");
			}
			ms_strcat(msgBuf, msg);
		}
		ms_strcat(msgBuf, "]");
		ms_strcat(msgBuf, "}");
		//ms_info("[%s]msgBuf:%s",pEVPGw_ctt->name,msgBuf);
		ms_s32 len_msg=ms_buflen(msgBuf);
		ms_s32 ret=pEVPGw_ctt->msptl_push->url_write(pEVPGw_ctt->pUrl_ctt,msgBuf,len_msg);
		//ms_logfile(FILE_LOG_WARING, "告警信息:%s,推送结果:%s,地址%s", msgBuf,(ret>0) ? "成功" : "失败",pEVPGw_ctt->pUrl_ctt->url);
		ms_waring("告警信息:%s,推送结果:%s,地址%s", msgBuf,(ret>0) ? "成功" : "失败",pEVPGw_ctt->pUrl_ctt->url);
		ms_memset0(pEvMsgList_ctt, ms_bufsize(MSEVENTMSGContext)*MAXNUM_MSG);
		numMsg=0;
		ms_deMalloc(msgBuf);
	}
	return numMsg;
}
static ms_bool msevp_innerapi_isSameEvent(MSEVENTMSGContext *pMsg1_ctt,MSEVENTMSGContext *pMsg2_ctt)
{
	return (ms_strncmp_snaeq(pMsg1_ctt->evType, pMsg2_ctt->evType)
			||pMsg1_ctt->evLevel!=pMsg2_ctt->evLevel
			||((ms_null!=pMsg1_ctt->devId)
				&&(ms_null!=pMsg2_ctt->devId)
				&&ms_strncmp_snaeq(pMsg1_ctt->devId, pMsg2_ctt->devId))
			) ? ms_false : ms_true;
}
static ms_void msevp_innerapi_eventPush(MSEVPGWContext *pEVPGw_ctt)
{
	msthread_api_setName(pEVPGw_ctt->name);
	MSFIFOLISTContext * *ppFifoListCtt_cp=&msEvpData_ctt.pFifoList_ctt;
	MSFIFOLISTContext *pFifiList_ctt=ms_null;
	MSEVENTMSGContext *pevMsg_ctt=ms_null;

	MSFIFOLISTContext *pFifiListList_ctt[MAXNUM_MSG];
	MSEVENTMSGFIFOContext msgFifo_ctt[eventPush_max];
	ms_u64 eventPush[eventPush_max]={0,1,5,10};//单位分钟
	while(1){
		if(ms_true==pEVPGw_ctt->flag_stop){
			goto msstop;
		}
	/*取出所有需要推送的消息:最多取出MAXNUM_MSG个消息,避免消息溢出*/
		ms_s32 num_msg=0;
		mslock_api_do(&msEvpData_ctt.mslock_ctt);
		ms_u32 totalnum=msfifo_api_listTotalNum(ppFifoListCtt_cp);
		while((totalnum>0)&&(num_msg<MAXNUM_MSG) ){
			pFifiListList_ctt[num_msg]=msfifo_api_listOut(ppFifoListCtt_cp);
			totalnum=msfifo_api_listTotalNum(ppFifoListCtt_cp);
			num_msg++;
		}
		mslock_api_undo(&msEvpData_ctt.mslock_ctt);

		//ms_debug("[%s]Get %d msg_in",pEVPGw_ctt->name,num_msg);
	/*合并相同的事件*/
		for(ms_s32 indexMsg=0;indexMsg<num_msg;indexMsg++){
			pFifiList_ctt=pFifiListList_ctt[indexMsg];
			if(ms_null!=pFifiList_ctt){
				pevMsg_ctt=pFifiList_ctt->ptr;
				MSEVENTMSGContext *pMsgList_ctt=msgFifo_ctt[pevMsg_ctt->timePush].msgList_ctt;
				if(msgFifo_ctt[pevMsg_ctt->timePush].numMsg<MAXNUM_MSG){
					ms_s32 index_realMsg=0;
					for(index_realMsg=0;index_realMsg<msgFifo_ctt[pevMsg_ctt->timePush].numMsg;index_realMsg++){
						if(ms_false==msevp_innerapi_isSameEvent(pevMsg_ctt,&pMsgList_ctt[index_realMsg])){
							continue;//相同消息，直接丢弃
						}else if(pMsgList_ctt[index_realMsg].inTime_us<pevMsg_ctt->inTime_us){
							pMsgList_ctt[index_realMsg].inTime_us=pevMsg_ctt->inTime_us;
							ms_strcpy(pMsgList_ctt[index_realMsg].jsSrc, pevMsg_ctt->jsSrc);
							break;//同类型消息，更新为最后一条报警
						}
					}
					if(!(index_realMsg<msgFifo_ctt[pevMsg_ctt->timePush].numMsg)){//新的事件,注册到时间FIFO中
						ms_memcpy(&pMsgList_ctt[index_realMsg], pevMsg_ctt, ms_bufsize(MSEVENTMSGContext));
						msgFifo_ctt[pevMsg_ctt->timePush].numMsg++;
					}
				/*如果消息数量过多，那么直接丢弃，避免内存溢出*/	
				}else{
					ms_waring("[%s]type(%d) num is %d,drop it",pEVPGw_ctt->name,pevMsg_ctt->timePush,msgFifo_ctt[pevMsg_ctt->timePush].numMsg);
				}
				msfifo_api_listDemalloc(&pFifiList_ctt);
			}
		}
		//ms_info("[%s]Get %d real Msg",pEVPGw_ctt->name,numRealMsg);
		//ms_info("[%s]Get %d real EmcMsg",pEVPGw_ctt->name,numRealMsgEmc);
		//构建数据集合，统一推送
		for(ms_s32 msgIndex=0;msgIndex<eventPush_max;msgIndex++){
			MSEVENTMSGFIFOContext *pMsgFifo_ctt=&msgFifo_ctt[msgIndex];
			if(ms_true==mstime_api_counterAsyncUs(&pMsgFifo_ctt->timePush_base, ms_usminutes(eventPush[msgIndex]))){
				pMsgFifo_ctt->numMsg=msevp_innerapi_msgPush(pEVPGw_ctt, pMsgFifo_ctt->numMsg,pMsgFifo_ctt->msgList_ctt);
			}	
		}
		ms_msleep(100);
	}
msstop:
	mslock_api_do(&msEvpData_ctt.mslock_ctt);
	ms_u32 totalnum=msfifo_api_listTotalNum(ppFifoListCtt_cp);
	mslock_api_undo(&msEvpData_ctt.mslock_ctt);
	while(totalnum>0){
		mslock_api_do(&msEvpData_ctt.mslock_ctt);
		pFifiList_ctt=msfifo_api_listOut(ppFifoListCtt_cp);
		totalnum=msfifo_api_listTotalNum(ppFifoListCtt_cp);
		mslock_api_undo(&msEvpData_ctt.mslock_ctt);
		if(ms_null!=pFifiList_ctt){
			msfifo_api_listDemalloc(&pFifiList_ctt);
		}
	}
	ms_debug("STOP============>%s",pEVPGw_ctt->name);
	pEVPGw_ctt->flag_stop=ms_false;
}

static ms_void msevp_innerapi_stateCheck(MSEVPGWContext *pEVPGw_ctt)
{
	msthread_api_setName(pEVPGw_ctt->name);

	ms_byte msg[2048]={0};
	msevp_innerapi_buildMsg(msg, mstime_api_us(), 
		eventLevel_ordinary_3, msevpType_str[eventType_heartBeat], MSEVP_DEFAULT_DEVID, ms_null);
	while(1){
		if(ms_true==pEVPGw_ctt->flag_stop){
			goto msstop;
		}
		ms_s32 len_msg=ms_buflen(msg);
		ms_s32 ret=pEVPGw_ctt->msptl_push->url_write(pEVPGw_ctt->pUrl_ctt,msg,len_msg);
		if(ret<0){
			pEVPGw_ctt->msptl_push->url_close(pEVPGw_ctt->pUrl_ctt);
			if(pEVPGw_ctt->msptl_push->url_open(pEVPGw_ctt->pUrl_ctt)<0){
				//ms_error("Open url(%s) failed",pEVPGw_ctt->pUrl_ctt->url);
			}else{
				ret=pEVPGw_ctt->msptl_push->url_write(pEVPGw_ctt->pUrl_ctt,msg,len_msg);
			}
			evp_ctt.flag_reach_pushUrl=ms_false;
		}else{
			evp_ctt.flag_reach_pushUrl=ms_true;
		}
		ms_sleep(10);
	}
msstop:
	ms_debug("STOP============>%s",pEVPGw_ctt->name);
	pEVPGw_ctt->flag_stop=ms_false;
}


ms_void msevp_api_eventIn(MSEVENTPUST_TIME timePush,MSEVENT_LEVEL evLevel,ms_string evType,ms_string devId,ms_string jsData)
{
//只有紧急告警才进行日志保存	
	if(evLevel<eventLevel_ordinary_1){
		ms_logfile(FILE_LOG_WARING, "告警信息:%s,推送情况:%s",jsData,(ms_false==msEvpData_ctt.flag_start) ? "未启用推送" : "正在推送");
	}
	if(ms_false==msEvpData_ctt.flag_start){
		ms_waring("告警信息:%s,推送情况:%s",jsData,"未启用推送");
	}else{
		//告警推送接口，建议放入FIFO，然后由专门的线程进行告警推送
		MSFIFOLISTContext *plist_ctt=msfifo_api_listMalloc(ms_bufsize(MSEVENTMSGContext), "EventMsg");
		MSEVENTMSGContext *pmsEvMsg_ctt=(MSEVENTMSGContext *)plist_ctt->ptr;
		pmsEvMsg_ctt->inTime_us=mstime_api_us();
		pmsEvMsg_ctt->timePush=timePush;
		pmsEvMsg_ctt->evLevel=evLevel;
		ms_strcpy(pmsEvMsg_ctt->evType, evType);
		ms_strcpy(pmsEvMsg_ctt->devId, devId);
		ms_strcpy(pmsEvMsg_ctt->jsSrc, jsData);
		mslock_api_do(&msEvpData_ctt.mslock_ctt);
		msfifo_api_listIn(&msEvpData_ctt.pFifoList_ctt,plist_ctt);
		mslock_api_undo(&msEvpData_ctt.mslock_ctt);
	} 
}
ms_void msevp_api_eventIn2(MSEVENT_TYPE evType,ms_string devId,ms_string jsData)
{
	MSEVENT_LEVEL eventLevel;
	MSEVENTPUST_TIME timePush=eventPush_nor;
	switch (evType){
		case eventType_authentException:
			eventLevel=eventLevel_emergency_3;
			timePush=eventPush_long;
			break;
		case eventType_nicException:		
		case eventType_cpuLoadException: 		
		case eventType_memException:			
		case eventType_diskException:
		case eventType_timeException:
			eventLevel=eventLevel_emergency_1;
			timePush=eventPush_short;
			break;
		case eventType_videoQualityMonitoring:			
			eventLevel=eventLevel_ordinary_1;
			timePush=eventPush_now;
			break;
		default:
			eventLevel=eventLevel_ordinary_3;
			timePush=eventPush_nor;
			break;
	}
	msevp_api_eventIn( timePush,eventLevel, msevpType_str[evType], devId,jsData);
}
ms_void msevp_api_createServer(ms_void)
{	
	MSEVENTPUSHContext *pEvp_ctt=&evp_ctt;
	if(0==ms_buflen(pEvp_ctt->pushUrl)){
		ms_waring("pushUrl is empty(%s),no need to start eventPush task",pEvp_ctt->pushUrl);
		msEvpData_ctt.flag_start=ms_false;
		return ;
	}else{
		ms_info("pushUrl is %s,start eventPush task",pEvp_ctt->pushUrl);
	}
	URLContext pushUrl_ctt;
	ms_memset0(&pushUrl_ctt, ms_bufsize(URLContext));
	//"ws://192.168.0.82:8989/"
	ms_strcpy(pushUrl_ctt.url, pEvp_ctt->pushUrl);
	msptc_api_urlSplit_urlctt(&pushUrl_ctt);
	URLProtocol *msptl_push=msptc_api_matchByName(pushUrl_ctt.url_info.ptcName);
	pushUrl_ctt.flags|=FLAG_WRITE;	
	pushUrl_ctt.opt.ws.dataType=WDT_TXTDATA;
	if(msptl_push->url_open(&pushUrl_ctt)<0){
		ms_error("Open url(%s) failed",pushUrl_ctt.url);
	}
	mslock_api_init(&msEvpData_ctt.mslock_ctt, "msEvpData_lock", lockType_mutex);
	MSEVPGWContext msEvpGw_ctt[MAXNUM_EVP_GW];
	MSTHREADContext msthreadGw_ctt[MAXNUM_EVP_GW];
	ms_u32 index=0;
	for(index=0;index<pEvp_ctt->numGW;index++){
		msEvpGw_ctt[index].msptl_push=msptl_push;
		msEvpGw_ctt[index].pUrl_ctt=&pushUrl_ctt;
		ms_sprintfs(msEvpGw_ctt[index].name, "evp_gw%d", index);
		msthread_api_create(&msthreadGw_ctt[index], msEvpGw_ctt[index].name,msevp_innerapi_eventPush,&msEvpGw_ctt[index]);
	}
	MSTHREADContext msthreadState_ctt;
	MSEVPGWContext msEvpState_ctt;
	msEvpState_ctt.msptl_push=msptl_push;
	msEvpState_ctt.pUrl_ctt=&pushUrl_ctt;
	ms_strcpy(msEvpState_ctt.name, "evp_stateCheck");
	msthread_api_create(&msthreadState_ctt, msEvpState_ctt.name,msevp_innerapi_stateCheck,&msEvpState_ctt);
	msEvpData_ctt.flag_start=ms_true;
}

#undef MSEVENT_PUSH_C

